package cn.com.itg.function.source;

import cn.com.itg.mapper.ZttcidenticalMapper;
import cn.com.itg.pojo.domain.entity.ZTTCIDENTICAL;
import cn.com.itg.util.SpringContextHolder;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.io.Serializable;
import java.net.CacheRequest;
import java.util.Calendar;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class ZttcidenticalSource extends RichSourceFunction<List<ZTTCIDENTICAL>>  implements Serializable {
    // 定时器间隔时间(ms)
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);
    private ZttcidenticalMapper zttcidenticalMapper;

//    public SignalSource(String sql) {
//        this.sql = sql;
//    }

    @Override
    public void open(Configuration parameters) throws Exception {
//        // 此处需要提前将mysql配置信息存储全局变量中 env.getConfig().setGlobalJobParameters(params);
//        ParameterTool params = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
//        String jdbcUrl = params.getRequired("jdbc.url");
//        String userName = params.getRequired("jdbc.username");
//        String password = params.getRequired("jdbc.password");
//        qr = new QueryRunner(JDBCUtils.getDataSource(jdbcUrl, userName, password));

        //这里注入mapper
        //通过spring 工厂读取配置文件
//        SpringContextHolder.setApplicationContext(new ClassPathXmlApplicationContext("applicationContext.xml"));
        zttcidenticalMapper = SpringContextHolder.getBean("zttcidenticalMapper");

    }

    @Override
    public void run(SourceContext<List<ZTTCIDENTICAL>> ctx) throws Exception {
//        while (true) {
//            List<ZTTCIDENTICAL> zttcidenticals = zttcidenticalMapper.selectList(
//                    Wrappers.<ZTTCIDENTICAL>query().lambda().orderByDesc(ZTTCIDENTICAL::getZdtype)
//            );
//            List<ZTTCIDENTICAL> collect = zttcidenticals.stream().map(s -> s.setTimestamp(Calendar.getInstance().getTime().getTime())).collect(Collectors.toList());
//            ctx.collect(collect);
//            // 每隔12小时执行一次
////            countDownLatch.await(1, TimeUnit.MINUTES);
//        }
        List<ZTTCIDENTICAL> zttcidenticals = zttcidenticalMapper.selectList(
                Wrappers.<ZTTCIDENTICAL>query().lambda().orderByDesc(ZTTCIDENTICAL::getZdtype)
        );
        List<ZTTCIDENTICAL> collect = zttcidenticals.stream().map(s -> s.setTimestamp(Calendar.getInstance().getTime().getTime())).collect(Collectors.toList());
        ctx.collect(collect);
    }

    @Override
    public void cancel() {

    }
}